﻿// StockTicker 
// - Microsoft StreamInsight application example
// - Contains input adapter for reading quotes from Yahoo Finance
// (C) Johan Åhlén, 2009

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Globalization;
using Microsoft.ComplexEventProcessing;
using Microsoft.ComplexEventProcessing.Adapters;

namespace AdvantIQ.StockTicker
{
    /// <summary>
    /// Simple input adapter that reads stock quotes
    /// Future development:
    /// - Check that the stock market is open
    /// - Check date/time of the source
    /// </summary>
    public class StockTickerTypedPointInput : TypedPointInputAdapter<StockPayload>
    {
        public readonly static IFormatProvider QuoteFormatProvider = CultureInfo.InvariantCulture.NumberFormat;
        private PointEvent<StockPayload> pendingEvent;
        private ScreenScraper screenScraper;
        private StockTickerInputConfig _config;

        /// <summary>
        /// Constructor
        /// </summary>
        /// <param name="config">Configuration for this adapter</param>
        public StockTickerTypedPointInput(StockTickerInputConfig config)
        {
            _config = config;
            YahooFinance tmp = new YahooFinance(config.StockSymbol);
            screenScraper = new ScreenScraper(tmp.URL,
                config.Timeout, tmp.MatchPattern);
        }

        public override void Start()
        {
            ProduceEvents();
        }

        public override void Resume()
        {
            ProduceEvents();
        }

        protected override void Dispose(bool disposing)
        {
            base.Dispose(disposing);
        }

        /// <summary>
        /// Main loop
        /// </summary>
        private void ProduceEvents()
        {
            var currEvent = default(PointEvent<StockPayload>);
            var ctr = 0;

            EnqueueCtiEvent(DateTime.Now);
            try
            {
                // Loop until stop signal
                while (AdapterState != AdapterState.Stopping)
                {
                    if (pendingEvent != null)
                    {
                        currEvent = pendingEvent;
                        pendingEvent = null;
                    }
                    else
                    {
                        // Read up to "NumberOfReadings" quotes
                        if (ctr++ < _config.NumberOfReadings)
                        {
                            try
                            {
                                // Read from source
                                var str = screenScraper.Scrape();
                                var value = double.Parse(str, QuoteFormatProvider);

                                // Produce INSERT event
                                currEvent = CreateInsertEvent();
                                currEvent.StartTime = DateTime.Now;
                                currEvent.Payload = new StockPayload { StockID = _config.StockName, Value = value };
                                pendingEvent = null;
                                //PrintEvent(currEvent);
                                Enqueue(ref currEvent);

                                // Also send an CTI event
                                EnqueueCtiEvent(DateTime.Now);

                            }
                            catch
                            {
                                // Error handling should go here
                            }
                            Thread.Sleep(_config.Interval);
                        }
                        else
                        {
                            break;
                        }
                    }
                }

                if (pendingEvent != null)
                {
                    currEvent = pendingEvent;
                    pendingEvent = null;
                }

                PrepareToStop(currEvent);
                Stopped();
            }
            catch (AdapterException e)
            {
                Console.WriteLine("AdvantIQ.StockTickerTypedPointInput.ProduceEvents - " + e.Message + e.StackTrace);
            }
        }

        private void PrepareToStop(PointEvent<StockPayload> currEvent)
        {
            //EnqueueCtiEvent(DateTime.Now);
            if (currEvent != null)
            {
                // Do this to avoid memory leaks
                ReleaseEvent(ref currEvent);
            }
        }

        private void PrepareToResume(PointEvent<StockPayload> currEvent)
        {
            pendingEvent = currEvent;
        }

        /// <summary>
        /// Debugging function
        /// </summary>
        /// <param name="evt"></param>
        private void PrintEvent(PointEvent<StockPayload> evt)
        {
            Console.WriteLine("Input: " + evt.EventKind + " " +
                evt.StartTime + " " + evt.Payload.StockID + " " +
                evt.Payload.Value);
        }
    }
}
